package cn.com.greatwall.kafka.service.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.servlet.http.HttpServletResponse;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import cn.com.greatwall.common.utils.FileUtil;
import cn.com.greatwall.common.utils.PageUtil;
import cn.com.greatwall.common.utils.QueryHelp;
import cn.com.greatwall.common.utils.ValidationUtil;
import cn.com.greatwall.kafka.domain.Task;
import cn.com.greatwall.kafka.repository.TaskRepository;
import cn.com.greatwall.kafka.service.ClusterService;
import cn.com.greatwall.kafka.service.KafkaService;
import cn.com.greatwall.kafka.service.TopicTaskService;
import cn.com.greatwall.kafka.service.dto.TaskDto;
import cn.com.greatwall.kafka.service.dto.criteria.TaskQueryCriteria;
import cn.com.greatwall.kafka.service.mapstruct.TaskMapper;
import cn.com.greatwall.kafka.util.KafkaUtil;
import cn.com.greatwall.system.modules.system.domain.Dept;
import lombok.RequiredArgsConstructor;

/**
 * @Author herw
 * @Time 2021-01-20 14:04:59
 * @Version 1.0
 * @Description: TODO(用一句话描述该文件做什么)
 */
@Service
@RequiredArgsConstructor
public class TopicTaskServiceImpl implements TopicTaskService {
    private final TaskRepository taskRepository;
    private final TaskMapper taskMapper;
    private final KafkaService kafkaService;
    private final ClusterService clusterService;

    @Override
    public void download(List<TaskDto> taskDtos, HttpServletResponse response) throws IOException {
        List<Map<String, Object>> list = new ArrayList<>();
        for (TaskDto taskDTO : taskDtos) {
            Map<String, Object> map = new LinkedHashMap<>();
            map.put("Topic Name", taskDTO.getName());
            map.put("Location", taskDTO.getLocation());
            map.put("Approval Status", taskDTO.getApproved());
            map.put("Operation", taskDTO.getApprovalOpinions());
            map.put("Comments", taskDTO.getComments());
            map.put("创建日期", taskDTO.getCreateTime());
            list.add(map);

        }
        FileUtil.downloadExcel(list, response);
    }
    
    @Override
    public Map<String, Object> queryAll(TaskQueryCriteria criteria, Pageable pageable) {
        Page<Task> page = taskRepository.findAll(
                (root, criteriaQuery, criteriaBuilder) -> QueryHelp.getPredicate(root, criteria, criteriaBuilder),
                pageable);
        return PageUtil.toPage(page.map(taskMapper::toDto));
    }

    @Override
    public List<TaskDto> queryAll(TaskQueryCriteria criteria) {
        return taskMapper.toDto(taskRepository.findAll(
                (root, criteriaQuery, criteriaBuilder) -> QueryHelp.getPredicate(root, criteria, criteriaBuilder)));
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void create(Task resources) {
        taskRepository.save(resources);
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void update(Task resources) {
        Task task = taskRepository.findById(resources.getId()).orElseGet(Task::new);
        ValidationUtil.isNull(task.getId(), "Task", "id", task.getId());
        resources.setId(task.getId());
        taskRepository.save(resources);
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void delete(Set<Long> ids) {
        taskRepository.deleteAllByIdIn(ids);
    }
    
    @Override
    public TaskDto findById(Long id) {
        Task task = taskRepository.findById(id).orElseGet(Task::new);
        ValidationUtil.isNull(task.getId(), "Task", "id", id);
        return taskMapper.toDto(task);
    }

    @Override
    public boolean checkTopicTask(TaskDto dto, String clusterIds) throws Exception {
        String[] clusterArray = clusterIds.split(",");
        for (String clusterId : clusterArray) {
            if(kafkaService.getKafkaUtil(clusterId).checkExists(dto.getName())) {
                throw new Exception(" Topic already exists in the" + clusterService.findById(Long.parseLong(clusterId)).getName() + " .");
            }
        }
        return false;
    }
}
